package com.alan.datasync.service.impl;

import com.alan.datasync.converter.OggJsonConverter;
import com.alan.datasync.service.IAbstractDataSyncService;
import com.alan.datasync.service.IOggKafkaMessageService;
import com.alan.datasync.wrapper.MessageEventWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.ParseException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: li_jl
 * @Description: 消息处理统一入口
 * @Date: Created in 2020/6/8 4:51 下午
 */
@Slf4j
@Service
public class OggKafkaMessageServiceImpl implements IOggKafkaMessageService {

    /**
     * 策略集合 key为表名
     */
    private final Map<String, IAbstractDataSyncService> dataSyncStrategy = new ConcurrentHashMap<>();

    @Autowired
    OggKafkaMessageServiceImpl(List<IAbstractDataSyncService> beans) {
        // 注入所有的实现类，以表名为key
        beans.forEach(v -> dataSyncStrategy.put(v.getTable(), v));
    }

    @Override
    public void dealMessage(String payload) {
        // 判空
        if (StringUtils.isBlank(payload)) {
            // 消息体为空，return
            return;
        }
        MessageEventWrapper messageEventWrapper = null;
        try {
            messageEventWrapper = OggJsonConverter.convert(payload);
        } catch (ParseException e) {
            // 报错等操作
        }

        // 判空
        if (StringUtils.isBlank(messageEventWrapper.getTable())) {
            // 表名都不明确，return
            return;
        }
        // 获取对应的策略
        IAbstractDataSyncService abstractDataSyncService = dataSyncStrategy.get(messageEventWrapper.getTable());
        if (null == abstractDataSyncService) {
            log.error("no dataSyncStrategy for this table : {}", messageEventWrapper.getTable());
            return;
        }

        switch (messageEventWrapper.getOpType()) {
            case DELETE:
                // 做同步删除处理
                abstractDataSyncService.deleteSync(messageEventWrapper);
                break;
            case INSERT:
                // 做同步插入处理
                abstractDataSyncService.insertSync(messageEventWrapper);
                break;
            case UPDATE:
                // 同步更新
                abstractDataSyncService.updateSync(messageEventWrapper);
                break;
            case UNKNOWN:
            default:
                break;
        }
    }
}
